package middleware.zookeeper;

import com.me.common.util.ConcurrentUtil;
import com.me.common.util.PropertiesUtil;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;

import static org.apache.zookeeper.Watcher.Event;

/**
 * zookeeper API：监听
 *
 * @author ME
 * @date 2018年8月20日 上午9:59:37
 */
public class TestZookeeper {
    private static final Logger LOGGER = LoggerFactory.getLogger(TestZookeeper.class);

    public static final String NODE_NAME_NODE_EXISTS = "/node-exists";
    public static final String NODE_NAME_GET_DATA = "/get-data";
    public static final String NODE_NAME_GET_CHILD = "/get-child";

    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(3, 5,
            2, TimeUnit.SECONDS, new SynchronousQueue<>(),
            new BasicThreadFactory.Builder().namingPattern("zk-pool-thread-%d").build());

    private static final String HOST_KEY = "zookeeper.host";

    public static void main(String[] args) throws InterruptedException {
        EXECUTOR.execute(TestZookeeper::existWatch);
        EXECUTOR.execute(TestZookeeper::getDataWatch);
        EXECUTOR.execute(TestZookeeper::getChildWatch);
    }

    private static void existWatch() {
        Thread.currentThread().setName("thread-exist-watch");

        try (ZooKeeperConnection connection = new ZooKeeperConnection()) {
            ZooKeeper zkClient = connection.connect(PropertiesUtil.getProperty(HOST_KEY));
            CountDownLatch latch = new CountDownLatch(1);

            Stat stat = zkClient.exists(NODE_NAME_NODE_EXISTS, event -> {
                System.out.println("EventType 变化时 KeeperState 都是 Event.KeeperState.SyncConnected");
                System.out.println("existWatch() 只能监听 Created、DataChanged、Deleted");
                // printEventTypeAndKeeperState(event);

                if (Event.EventType.NodeCreated == event.getType()
                        || Event.EventType.NodeDataChanged == event.getType()) {
                    String newData = ZkUtil.getData(zkClient, NODE_NAME_NODE_EXISTS);
                    System.out.println(NODE_NAME_NODE_EXISTS + " 新值：" + newData);
                } else if (event.getType() == Event.EventType.NodeDeleted) {
                }

                latch.countDown();
            });

            if (Objects.nonNull(stat)) {
                System.out.println("Stat：" + stat);
            } else {
                System.out.println("元素不存在：" + NODE_NAME_NODE_EXISTS);
            }
            System.out.println();

            ConcurrentUtil.latchAwait(latch);
            System.out.println();
        } catch (Exception e) {
            LOGGER.info(e.getMessage(), e);
        }
    }

    private static void getDataWatch() {
        Thread.currentThread().setName("thread-data-watch");

        try (ZooKeeperConnection connection = new ZooKeeperConnection()) {
            ZooKeeper zkClient = connection.connect(PropertiesUtil.getProperty(HOST_KEY));
            ZkUtil.create(zkClient, NODE_NAME_GET_DATA, "old", CreateMode.EPHEMERAL);

            CountDownLatch latch = new CountDownLatch(1);

            byte[] b = zkClient.getData(NODE_NAME_GET_DATA, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    System.out.println("EventType 变化时 KeeperState 都是 Event.KeeperState.SyncConnected");
                    System.out.println("getData() 只能监听 DataChanged、Deleted");
                    // printEventTypeAndKeeperState(event);

                    if (Event.EventType.NodeDataChanged == event.getType()) {
                        String newData = ZkUtil.getData(zkClient, NODE_NAME_GET_DATA);
                        System.out.println(NODE_NAME_GET_DATA + " 新值：" + newData);
                    } else if (event.getType() == Event.EventType.NodeDeleted) {
                    }

                    latch.countDown();
                }
            }, null);

            String data = new String(b, StandardCharsets.UTF_8);
            System.out.println(NODE_NAME_GET_DATA + " 旧值：" + data);

            ConcurrentUtil.latchAwait(latch);
            System.out.println();
            ZkUtil.delete(zkClient, NODE_NAME_GET_DATA);
        } catch (Exception e) {
            LOGGER.info(e.getMessage(), e);
        }
    }

    private static void getChildWatch() {
        Thread.currentThread().setName("thread-child-watch");

        try (ZooKeeperConnection connection = new ZooKeeperConnection()) {
            ZooKeeper zkClient = connection.connect(PropertiesUtil.getProperty(HOST_KEY));
            ZkUtil.create(zkClient, NODE_NAME_GET_CHILD, "old", CreateMode.PERSISTENT);

            CountDownLatch latch = new CountDownLatch(1);

            List<String> childList = zkClient.getChildren(NODE_NAME_GET_CHILD, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    System.out.println("EventType 变化时 KeeperState 都是 Event.KeeperState.SyncConnected");
                    System.out.println("getChildren() 只能监听 Deleted、ChildrenChanged");
                    printEventTypeAndKeeperState(event);

                    if (Event.EventType.NodeChildrenChanged == event.getType()) {
                        List<String> childList = ZkUtil.getChild(zkClient, NODE_NAME_GET_CHILD);
                        System.out.println("变更后的子节点列表：" + childList);
                    } else if (event.getType() == Event.EventType.NodeDeleted) {

                    }

                    latch.countDown();
                }
            }, null);

            System.out.println("变更前的子节点列表：" + childList);
            ConcurrentUtil.latchAwait(latch);
            System.out.println();

            ZkUtil.delete(zkClient, NODE_NAME_GET_CHILD);
        } catch (Exception e) {
            LOGGER.info(e.getMessage(), e);
        }
    }

    private static void printEventTypeAndKeeperState(WatchedEvent event) {
        System.out.println("当前 EventType：" + event.getType());
        System.out.println("EventType 对应的 KeeperState：" + event.getState());
        System.out.println("当前 path：" + event.getPath());
    }

    /**
     * 给所有节点递归生成 setAcl 命令
     * 
     * @param path 父路径
     * @param ip 权限信息
     * @throws KeeperException
     * @throws InterruptedException
     */
    public static void printSetAcl(String path, String ip) {
        try (ZooKeeperConnection connection = new ZooKeeperConnection()) {
            ZooKeeper zkClient = connection.connect(PropertiesUtil.getProperty(HOST_KEY));
            List<String> children = zkClient.getChildren(path, false);
            for (String str : children) {
                if ("/".equals(path)) {
                    System.out.println("setAcl " + path + str + " " + ip);
                    printSetAcl(path + str, ip);
                } else {
                    System.out.println("setAcl " + path + "/" + str + " " + ip);
                    printSetAcl(path + "/" + str, ip);
                }
            }
        } catch (Exception e) {
            LOGGER.info(e.getMessage(), e);
        }
    }

}
